package com.aispeech.task.mgr.task;

import com.aispeech.constant.RedisKeys;
import com.aispeech.data.pojo.task.Task;
import com.aispeech.redis.DbMgr;
import com.aispeech.task.mgr.event.TaskEvent;
import com.aispeech.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

import java.util.List;
import java.util.NoSuchElementException;

@Slf4j
public class EventSource extends RichParallelSourceFunction<TaskEvent> {
    private volatile boolean running = true;
    private DbMgr dbMgr = DbMgr.Instance();

    @Override
    public void open(Configuration params) throws Exception {
        super.open(params);

        if (!dbMgr.Initialized()) init();
    }

    private void init() {
        dbMgr.initialize("redis://10.12.8.30/3");
    }

    protected TaskEvent newEvent(Task task) { throw new NotImplementedException(); }

    @Override
    public void run(SourceContext<TaskEvent> ctx) throws Exception {
        while (running) {
			String productIdAndTaskId;

        	try {
				List<String> tasksInWip = dbMgr.zrange(RedisKeys.SORTEDSET_TASKs, 0, 1);
				if (tasksInWip.size() > 0) {
					productIdAndTaskId = tasksInWip.get(0);
				} else {
					productIdAndTaskId = "";
				}

//				productIdAndTaskId = dbMgr.zpopmax(RedisKeys.SORTEDSET_TASKs, String.class);
			} catch (NoSuchElementException e) {
				productIdAndTaskId = "";
			}

			if (!StringUtils.isNullOrEmpty(productIdAndTaskId)) {
				int lastDotIdx = productIdAndTaskId.lastIndexOf(".");
				String productId = productIdAndTaskId.substring(0, lastDotIdx),
					taskId = productIdAndTaskId.substring(lastDotIdx + 1);

				Task task = dbMgr.hget(RedisKeys.HASH_TASK("task_", productId), taskId, Task.class);

				if (task != null) {
					ctx.collect(newEvent(task));
				} else {
					log.error("Not found task info. productId: {}, taskId: {}", productId, taskId);
				}
			}


            Thread.sleep(3000L);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
